package study.mq.kafka.consumer.listener._3_batch;

import cn.hutool.core.thread.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * 批量拉取
 * （1）consumer.type=BATCH
 * （2）consumer.max-poll-records=100
 */
@Slf4j
@Component
public class Batch_01_Listener {

   // @KafkaListener(topics = "topic-3")
    public void onMessage(ConsumerRecords<String, String> records) {
        log.info(">>>>>> count: {}", records.count());
        //
        records.partitions().forEach(tp -> {
            String topic = tp.topic();
            int partition = tp.partition();
            log.info("topic= {}, partition= {}", topic, partition);
        });
        //
        records.forEach(record -> {
            ThreadUtil.sleep(1 * 1000);
            //
            String topic = record.topic();
            int partition = record.partition();
            long offset = record.offset();
            log.info("[{}, {}] offset= {}", topic, partition, offset);
        });
    }
}
